package org.apache.solr.ocean.search;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

import org.apache.solr.ocean.cache.DocSetCacheKey;
import org.apache.solr.ocean.core.Snapshot;
import org.apache.solr.ocean.core.Index.IndexSnapshot;
import org.apache.solr.ocean.util.Timeout;
import org.apache.solr.search.DocSet;
import org.apache.solr.search.MultiDocSet;

/**
 * Obtains doc sets for a snapshot
 *
 */
public class SnapshotDocSetSearch implements Callable<DocSet> {
  private Snapshot snapshot;
  private ExecutorService executorService;
  private Timeout timeout;
  private DocSetCacheKey docSetCacheKey;
  
  public SnapshotDocSetSearch(DocSetCacheKey docSetCacheKey, Snapshot snapshot, Timeout timeout, ExecutorService executorService) {
    this.docSetCacheKey = docSetCacheKey;
    this.snapshot = snapshot;
    this.timeout = timeout;
    this.executorService = executorService;
  }
  
  public DocSet call() throws Exception {
    List<IndexSnapshot> indexSnapshots = snapshot.getIndexSnapshots();
    List<Callable<DocSetSearchResult>> indexSearches = new ArrayList<Callable<DocSetSearchResult>>(indexSnapshots.size());
    for (int x=0; x < indexSnapshots.size(); x++) {
      IndexSnapshot indexSnapshot = indexSnapshots.get(x);
      indexSearches.add(new IndexSnapshotDocSetSearch(docSetCacheKey, x, indexSnapshot, timeout));
    }
    List<Future<DocSetSearchResult>> futures = executorService.invokeAll(indexSearches);
    List<DocSetSearchResult> results = new ArrayList<DocSetSearchResult>(indexSnapshots.size());
    for (Future<DocSetSearchResult> future : futures) {
      DocSetSearchResult result = future.get();
      results.add(result);
    }
    Collections.sort(results);
    DocSet[] docSets = new DocSet[results.size()];
    long[] indexIds = new long[results.size()];
    for (int x=0; x < results.size(); x++) {
      DocSetSearchResult result = results.get(x);
      docSets[x] = result.docSet;
      indexIds[x] = result.indexSnapshot.getIndex().getId();
    }
    int[] starts = snapshot.getStarts();
    return new SnapshotDocSet(snapshot.getId(), docSets, starts, indexIds, docSetCacheKey);
  }
  
  private static class DocSetSearchResult implements Comparable<DocSetSearchResult> {
    public int index;
    public IndexSnapshot indexSnapshot;
    public DocSet docSet;
    
    public DocSetSearchResult(int index, IndexSnapshot indexSnapshot, DocSet docSet) {
      this.index = index;
      this.indexSnapshot = indexSnapshot;
      this.docSet = docSet;
    }

    public int compareTo(DocSetSearchResult other) {
      if (index > other.index) return 1;
      if (index < other.index) return -1;
      return 0;
    }
  }
  
  private static class IndexSnapshotDocSetSearch implements Callable<DocSetSearchResult> {
    private int index;
    private IndexSnapshot indexSnapshot;
    private DocSetCacheKey docSetCacheKey;
    private Timeout timeout;
    
    private IndexSnapshotDocSetSearch(DocSetCacheKey docSetCacheKey, int index, IndexSnapshot indexSnapshot, Timeout timeout) {
      this.docSetCacheKey = docSetCacheKey;
      this.index = index;
      this.indexSnapshot = indexSnapshot;
      this.timeout = timeout;
    }
    
    public DocSetSearchResult call() throws Exception {
      DocSet docSet = indexSnapshot.getDocSet(docSetCacheKey, timeout);
      return new DocSetSearchResult(index, indexSnapshot, docSet);
    }
  }
}
